package main;
/*
 * CloudStego school project
 * @author lauris kruusamae
 * @author triin uudam
 * @author uku loskit
 */
import java.awt.image.BufferedImage;
import java.io.IOException;
import java.util.ArrayList;

import javax.imageio.ImageIO;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.log4j.Logger;

/**
 * Reducer implementation for message decoding
 * key - image name
 * value - image path
 * 
 * Extract the message and saves it to output directory
 * with <image name>-message.png as its path
 *
 */
class DecodeReduce extends Reducer<Text, Text, IntWritable, IntWritable> {
    
    public static final String KEY_OUTPUT_DIR_PATH = "output_dir_path";
    
    private String resultDirPath;
    Logger logger = Logger.getLogger(DecodeReduce.class);
    
    /** Reducer setup */
    public void setup (Context context) throws IOException{
    	logger.info("SETUP CALLED");
        resultDirPath = context.getConfiguration().get(KEY_OUTPUT_DIR_PATH);
        if (resultDirPath.isEmpty()) {
            throw new IllegalArgumentException("resultPath cannot be null");
        }
    }

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException, NumberFormatException {
    	FileSystem fs = FileSystem.get(context.getConfiguration());
    	
    	for (Text value : values) {
    		FSDataInputStream in = fs.open(new Path(value.toString()));
    		BufferedImage baseImage = ImageIO.read(in);
    		in.close();
    		ArrayList<Pixel> basePixels = Util.getPixelsFromImage(baseImage, false);
    		
    		Path messagePath = new Path(resultDirPath + "/" + Util.getFileName(value.toString()) + "-message.png");
    		FSDataOutputStream out = fs.create(messagePath);
    		Util.extractMessage(out, basePixels);
    		out.flush();
    		out.close();
    	}
    }
}
